package com.google.code.fluentjoiner.example;

import com.google.code.fluentjoiner.Fold;
import com.google.code.fluentjoiner.IteratorSupplierBuilder;
import com.google.code.fluentjoiner.io.local.LocalFileObjectSupplierBuilder;
import com.google.common.base.Function;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMultiset;
import com.google.common.collect.Multiset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import static com.google.code.fluentjoiner.StringFunctions.*;
import java.util.Iterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Netflix
{
    private static final Logger LOGGER = LoggerFactory.getLogger(Netflix.class);
    public static void main(final String[] args)
    {
        final ExecutorService executorService = Executors.newFixedThreadPool(5);
        final Multiset<Integer> overallRatings = LocalFileObjectSupplierBuilder.filesInFolder(System.getProperty("netflix.location"))
                .parallelElementTransform(new Function<LocalFileObjectSupplierBuilder, Multiset<Integer>>()
                {
                    @Override
                    public Multiset<Integer> apply(final LocalFileObjectSupplierBuilder localFileObjectSupplierBuilder)
                    {
                        final Iterator<String> lines = localFileObjectSupplierBuilder.textFromFile()
                                .map(split("\n"))
                                .get();
                        lines.next();
                        final ImmutableList<Integer> ratings = IteratorSupplierBuilder.fromIterator(lines)
                                .elementTransform(new Function<String, Integer>()
                                {
                                    @Override
                                    public Integer apply(final String line)
                                    {
                                        return Integer.parseInt(line.split(",")[1]);
                                    }
                                })
                                .toImmutableList()
                                .get();
                         return ImmutableMultiset.copyOf(ratings);
                    }
                }, executorService)
                .foldLeft(new Fold<Multiset<Integer>, Multiset<Integer>>()
                {
                    @Override
                    public Multiset<Integer> apply(final Multiset<Integer> from, final Multiset<Integer> priorValue)
                    {
                        final ImmutableMultiset.Builder<Integer> builder = ImmutableMultiset.builder();
                        builder.addAll(from);
                        builder.addAll(priorValue);
                        return builder.build();
                    }
                }, new ImmutableMultiset.Builder<Integer>().build())
                .get();
        LOGGER.info("About to shutdown the ExecutorService.");
        executorService.shutdown();
        LOGGER.info("Results = {}", overallRatings);
    }
}
